package com.horizon.im.imServer.distributed;

import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import com.horizon.im.common.bean.msg.ProtoMsg;
import com.horizon.im.imServer.protoBuilder.NotificationMsgBuilder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;

import com.horizon.constants.ServerConstants;
import com.horizon.entity.ImNode;
import com.horizon.util.JsonUtil;
import com.horizon.util.ObjectUtil;
import com.horizon.zk.CuratorZKClient;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * 如果连接在不同的Netty Worker工作站点的客户端之间需要相互进行消息的发送， 就需要在不同的Worker节点之间进行路由和转发。 Worker节点的路由是指根据消息需要转发的目标用户找到用户连接所在的Worker节点。
 * 由于节点和节点之间都有可能需要相互转发，因此节点之间的连接是一种网状结构。 每一个节点都需要具备路由的能力。
 */
@Slf4j
@Data
public class WorkerRouter {
    /**
     * 为每一个Worker节点增加一个IM路由器类，名为WorkerRouter。 为了能够转发到所有的节点， 一是要订阅到集群中所有的在线Netty服务器，并且保存起来； 二是要其他的Netty服务器建立一个长连接，用于转
     */
    // zk客户端
    private CuratorFramework client = null;
    private String pathRegistered = null;
    private ImNode imNode = null;
    private static WorkerRouter singetonInstance = null;
    private static final String path = ServerConstants.MANAGE_PATH;

    private ConcurrentHashMap<Long, PeerSender> workerMap = new ConcurrentHashMap<Long, PeerSender>();

    private BiConsumer<ImNode, PeerSender> runAfterAdd = (node, relaySender) -> {
        doAfterAdd(node, relaySender);
    };

    private Consumer<ImNode> runAfterRemove = (node) -> {
        doAfterRemove(node);
    };

    static class SingletonInstance {
        private static WorkerRouter workerRouter = new WorkerRouter();
    }

    public static WorkerRouter getInstance() {
        return SingletonInstance.workerRouter;
    }

    private boolean inited = false;

    // 初始化节点管理
    public void init() {
        if (inited) {
            return;
        }

        inited = true;
        if (null == client) {
            this.client = CuratorZKClient.getInstance().getClient();
        }

        // 订阅节点的增加和删除事件
        PathChildrenCache childrenCache = new PathChildrenCache(client, path, true);
        PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent)
                throws Exception {
                log.info("开始监听其他的ImWorker子节点:-----");
                ChildData data = pathChildrenCacheEvent.getData();
                switch (pathChildrenCacheEvent.getType()) {
                    case CHILD_ADDED:
                        log.info("CHILD_ADDED : " + data.getPath() + "  数据:" + data.getData());
                        processNodeAdded(data);
                        break;
                    case CHILD_REMOVED:
                        log.info("CHILD_REMOVED : " + data.getPath() + "  数据:" + data.getData());
                        processNodeRemoved(data);
                        break;
                    case CHILD_UPDATED:
                        log.info("CHILD_UPDATED : " + data.getPath() + "  数据:" + new String(data.getData()));
                        break;
                    default:
                        log.debug("[PathChildrenCache]节点数据为空, path={}", data == null ? "null" : data.getPath());
                        break;
                }
            }
        };

        childrenCache.getListenable().addListener(childrenCacheListener);
        try {
            childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // ProcessNodeRemoved
    private void processNodeRemoved(ChildData data) {
        byte[] payload = data.getData();
        ImNode node = ObjectUtil.JsonBytes2Object(payload, ImNode.class);

        long id = ImWorker.getInstance().getIdByPath(data.getPath());
        node.setId(id);
        log.info("[TreeCache]节点删除，path={}, data={}", data.getPath(), JsonUtil.pojoToJson(node));
        // 从workerMap中获取peerSender对端信息，本端保存了一个和对端的node
        // 相关联的peerSender对象，用于保存和哪个对端建立了路由信息，
        // 当当前节点下线后，需要查找之前连接的对端节点，关闭EventLoop
        PeerSender peerSender = workerMap.get(node.getId());

        if (null != peerSender) {
            // 直接关闭对应的eventloop即可
            peerSender.stopConnecting();
            workerMap.remove(node.getId());
        }
    }

    private void processNodeAdded(ChildData data) {
        byte[] payload = data.getData();
        ImNode node = ObjectUtil.JsonBytes2Object(payload, ImNode.class);
        
        long id = ImWorker.getInstance().getIdByPath(data.getPath());
        node.setId(id);
        log.info("[TreeCache]节点更新端口，path={}, data={}", data.getPath(), JsonUtil.pojoToJson(node));
        
        if (node.equals(getLocalNode())){
            log.info("[PathChildrenCach]添加的是本地节点：path={},data={};",data.getPath(), JsonUtil.pojoToJson(node));
            return;
        }

        PeerSender relaySender = workerMap.get(node.getId());
        //重复收到注册事件
        if (null != relaySender && relaySender.getRemoteImNode().equals(node)){
            log.info("[PathChildrenCache]节点重复增加，path={}, data={}", data.getPath(), JsonUtil.pojoToJson(node));
            return;
        }

        if (runAfterAdd != null) {
            runAfterAdd.accept(node, relaySender);
        }
    }

    public ImNode getLocalNode() {
        return ImWorker.getInstance().getLocalNode();
    }

    private void doAfterRemove(ImNode node) {
        PeerSender peerSender = workerMap.get(node.getId());

        if (null != peerSender) {
            peerSender.stopConnecting();
            workerMap.remove(node.getId());
        }


    }

    private void doAfterAdd(ImNode n, PeerSender relaySender) {
        if (null != relaySender) {
            //关闭老的连接
            relaySender.stopConnecting();
        }
        //创建一个消息转发器
        relaySender = new PeerSender(n);
        //建立转发的连接
        relaySender.doConnect();

        workerMap.put(n.getId(), relaySender);
    }

    public PeerSender route(long nodeId) {
        PeerSender peerSender = workerMap.get(nodeId);
        if (null != peerSender) {
            return peerSender;
        }
        return null;
    }


    public void sendNotification(String json) {
        workerMap.keySet().stream().forEach(
                key ->
                {
                    if (!key.equals(getLocalNode().getId())) {
                        PeerSender peerSender = workerMap.get(key);
                        ProtoMsg.Message pkg = NotificationMsgBuilder.buildNotification(json);
                        peerSender.writeAndFlush(pkg);
                    }
                }
        );

    }
}
